package com.yoome.easyrpc.common.utils;

import com.yoome.easyrpc.common.enums.ErrorCode;
import com.yoome.easyrpc.common.properties.ZkEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class CuratorUtil {

    private static String zkAddress;
    private static Integer baseSleepTime;
    private static Integer maxRetries;
    private static Integer maxWaitTime;
    private static CuratorFramework zkClient;

    public static void initProperties(ZkEntity entity)
    {
        zkAddress = entity.getZkAddress();
        baseSleepTime = entity.getBaseSleepTime();
        maxRetries = entity.getMaxRetries();
        maxWaitTime = entity.getMaxWaitTime();
    }
//
//    @Value("${zookeeper.zkAddress:127.0.0.1:2181}")
//    public static void setZkAddress(String zkAddress) {
//        CuratorUtil.zkAddress = zkAddress;
//    }
//
//    @Value("${zookeeper.baseSleepTime:1000}")
//    public static void setBaseSleepTime(Integer baseSleepTime) {
//        CuratorUtil.baseSleepTime = baseSleepTime;
//    }
//
//    @Value("${zookeeper.maxRetries:3}")
//    public static void setMaxRetries(Integer maxRetries) {
//        CuratorUtil.maxRetries = maxRetries;
//    }
//
//    @Value("${zookeeper.maxWaitTime:30}")
//    public static void setMaxWaitTime(Integer maxWaitTime) {
//        CuratorUtil.maxWaitTime = maxWaitTime;
//    }

    public static final Map<String, List<String>> SERVICE_MAP = new ConcurrentHashMap<>();
    private static final Set<String> REGISTERED_PATHS = ConcurrentHashMap.newKeySet();
    public static final String ZK_ROOT_PATH = "/easy-rpc";
    

    public static CuratorFramework getClient() {
        if (Objects.nonNull(zkClient) && CuratorFrameworkState.STARTED.equals(zkClient.getState())) {
            return zkClient;
        }

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTime, maxRetries);
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkAddress)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        try {
            if (!zkClient.blockUntilConnected(maxWaitTime, TimeUnit.SECONDS)) {
                throw new RuntimeException("not connect to zookeeper!");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return zkClient;
    }


    public static List<String> getChildNodes(CuratorFramework client, String serviceName) {
        if (SERVICE_MAP.containsKey(serviceName)) {
            return SERVICE_MAP.get(serviceName);
        }

        String path = ZK_ROOT_PATH + "/" + serviceName;
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        PathChildrenCacheListener listener = (curator, event) -> {
            List<String> servicePaths = curator.getChildren().forPath(path);
            SERVICE_MAP.put(serviceName, servicePaths);
        };
        cache.getListenable().addListener(listener);
        try {
            List<String> servicePaths = client.getChildren().forPath(path);
            SERVICE_MAP.put(serviceName, servicePaths);
            cache.start();
        } catch (Exception e) {
            log.info("zk watcher start error!", e);
            throw new EasyRpcException(ErrorCode.SERVICE_NOT_FOUND);
        }
        return SERVICE_MAP.get(serviceName);
    }


    public static void createPersistentNode(CuratorFramework client, String path) {
        try {
            if (!REGISTERED_PATHS.contains(path) && Objects.isNull(client.checkExists().forPath(path))) {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                REGISTERED_PATHS.add(path);
                log.info("create persistent zk node with path:{}", path);
            }
        } catch (Exception e) {
            log.error("create persistent zk node error!", e);
        }
    }
}
